BigQuery テーブル更新トリガで、Dataplex のプロファイリングを実行してみた。

gcloud iam service-accounts create sa-functions \
    --display-name "Functions実行用サービスアカウント"

gcloud projects add-iam-policy-binding [PROJECT_ID] \
    --member="serviceAccount:sa-functions@[PROJECT_ID].iam.gserviceaccount.com" \
    --role="roles/run.invoker"
gcloud projects add-iam-policy-binding [PROJECT_ID] \
    --member="serviceAccount:sa-functions@[PROJECT_ID].iam.gserviceaccount.com" \
    --role="roles/eventarc.eventReceiver"
gcloud projects add-iam-policy-binding [PROJECT_ID] \
    --member="serviceAccount:sa-functions@[PROJECT_ID].iam.gserviceaccount.com" \
    --role="roles/dataplex.admin"
gcloud projects add-iam-policy-binding [PROJECT_ID] \
    --member="serviceAccount:sa-functions@[PROJECT_ID].iam.gserviceaccount.com" \
    --role="roles/bigquery.admin"

from google.cloud import dataplex_v1
import functions_framework
import os
import time

def run_data_scan(project_id, location, data_scan_id, data_source_entity):
    client = dataplex_v1.DataScanServiceClient()

    parent = f"projects/{project_id}/locations/{location}"
    data_scan_name = f"{parent}/dataScans/{data_scan_id}"

    # データスキャンジョブが既に存在するか確認
    existing_scans = client.list_data_scans(parent=parent)
    scan_exists = any(scan.name == data_scan_name for scan in existing_scans)

    if not scan_exists:
        # 同名のジョブが存在しない場合はデータスキャンジョブを作成
        data_scan = dataplex_v1.DataScan(
            data=dataplex_v1.DataSource(
                entity=data_source_entity
            ),
            data_profile_spec=dataplex_v1.DataProfileSpec(
                sampling_percent=10
            )
        )
        request = dataplex_v1.CreateDataScanRequest(
            parent=parent,
            data_scan_id=data_scan_id,
            data_scan=data_scan
        )
        operation = client.create_data_scan(request=request)
        print("[functions] Waiting for operation to complete...")
        response = operation.result()
        print(f"[functions] Data scan created: {response.name}")
    else:
        print(f"[functions] Data scan with ID '{data_scan_id}' already exists.")

    # データスキャンジョブを実行
    run_request = dataplex_v1.RunDataScanRequest(name=data_scan_name)
    run_response = client.run_data_scan(request=run_request)
    name = run_response.job.name
    print(f"[functions] Data scan job started: {name}")

    # データスキャンジョブのステータスをポーリングして確認
    while True:
        job = client.get_data_scan_job(name=name)
        if job.state == dataplex_v1.DataScanJob.State.SUCCEEDED:
            return True
        elif job.state == dataplex_v1.DataScanJob.State.FAILED:
            print(f"[functions] error: {job.error.message}")
            return False
        else:
            print("[functions] Waiting for data scan to complete...")
            time.sleep(30)  # 30秒待機

@functions_framework.cloud_event
def main(cloudevent):
    print(f"Event type: {cloudevent['type']}")
    if 'subject' in cloudevent:
        print(f"Subject: {cloudevent['subject']}")

    # 環境変数チェック
    PROJECT_ID = os.getenv('PROJECT_ID')
    SA_DATAFORM = os.getenv('SA_DATAFORM')
    LOCATION = os.getenv('LOCATION')
    LAKE_ID = os.getenv('LAKE_ID')
    ZONE_ID = os.getenv('ZONE_ID')
    if not PROJECT_ID or not SA_DATAFORM or not LOCATION or not LAKE_ID or not ZONE_ID:
        raise Exception('Invalid env val.')    

    # イベントデータチェック
    payload = cloudevent.data.get("protoPayload")
    if not payload:
        return
    principalEmail = payload.get('authenticationInfo', {}).get('principalEmail', None)
    if principalEmail != SA_DATAFORM:
        # Dataformからのクエリじゃない場合は処理せず終了
        return
    jobChange = payload.get('metadata', {}).get('jobChange', None)
    if not jobChange:
        return
    statementType = jobChange.get('job', {}).get('jobConfig', {}).get('queryConfig', {}).get('statementType', None)
    if statementType != 'CREATE_TABLE_AS_SELECT':
        return
    statementType = jobChange.get('job', {}).get('jobConfig', {}).get('queryConfig', {}).get('statementType', None)
    destinationTable = jobChange.get('job', {}).get('jobConfig', {}).get('queryConfig', {}).get('destinationTable', None)
    dst_dataset = destinationTable.split('/')[3]
    dst_table = destinationTable.split('/')[5]
    print(f'[functions] dst_dataset: {dst_dataset} / dst_table: {dst_table}')

    entity_id = dst_table
    data_scan_id = f"scan-{entity_id.replace('_', '-')}"
    data_source_entity = f"projects/{PROJECT_ID}/locations/{LOCATION}/lakes/{LAKE_ID}/zones/{ZONE_ID}/entities/{entity_id}"    

    run_result = run_data_scan(PROJECT_ID, LOCATION, data_scan_id, data_source_entity)
    if run_result:
        print("[functions] Data scan completed successfully.")
    else:
        print(f"[functions] Data scan failed...")

functions-framework==3.*
google-cloud-dataplex>=2.2.2

gcloud functions deploy exec_profiling \
    --gen2 \
    --region asia-northeast1 \
    --runtime python312 \
    --entry-point=main \
    --trigger-event-filters="type=google.cloud.audit.log.v1.written" \
    --trigger-event-filters="serviceName=bigquery.googleapis.com" \
    --trigger-event-filters="methodName=google.cloud.bigquery.v2.JobService.InsertJob" \
    --trigger-location=asia-northeast1 \
    --service-account=sa-functions@[PROJECT_ID].iam.gserviceaccount.com \
    --trigger-service-account=sa-functions@[PROJECT_ID].iam.gserviceaccount.com \
    --set-env-vars PROJECT_ID=[PROJECT_ID],SA_DATAFORM=service-[PROJECT_NO]@gcp-sa-dataform.iam.gserviceaccount.com,LOCATION=asia-northeast1,LAKE_ID=lake-tokyo,ZONE_ID=marketing

（省略）
ERROR: (gcloud.functions.deploy) OperationError: code=3, message=Could not create or update Cloud Run service exec-profiling, Container Healthcheck failed. Revision 'exec-profiling-00003-pid' is not ready and cannot serve traffic. The user-provided container failed to start and listen on the port defined provided by the PORT=8080 environment variable. Logs for this revision might contain more information.
（省略）
For more troubleshooting guidance, see https://cloud.google.com/run/docs/troubleshooting#container-failed-to-start

google.api_core.exceptions.PermissionDenied: 403 Permission denied for user to access source BigQuery table 'mart_test'. You need to have `bigquery.tables.get` permission for the specified table to create a DataScan for it."

from google.cloud import bigquery

def add_label_table(project_id, location, data_scan_id, dataset_id, table_id):
    label_key_scan = "dataplex-dp-published-scan"
    label_key_project = "dataplex-dp-published-project"
    label_key_location = "dataplex-dp-published-location"

    client = bigquery.Client()

    dataset_ref = bigquery.DatasetReference(project_id, dataset_id)
    table_ref = dataset_ref.table(table_id)
    table = client.get_table(table_ref)
    print(table.labels)

    if label_key_scan in table.labels and label_key_project in table.labels and label_key_location in table.labels:
        print('Label is already attached.')
        return

    table.labels = {
        label_key_scan: f"{data_scan_id}",
        label_key_project: f"{project_id}",
        label_key_location: f"{location}"
    }
    table = client.update_table(table, ["labels"])

    print("Labels added to {}".format(table_id))

（省略）

def main(cloudevent):
（省略）
    add_label_table(PROJECT_ID, LOCATION, data_scan_id, dst_dataset, dst_table)